package com.wei.mutualfriend;

import java.io.IOException;
import java.util.Arrays;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.Reducer;


/**
 * @author Administrator
 */
public class MutualFriendStepTwo {
	static class MutualFriendMap extends Mapper<LongWritable, Text, Text, Text>{
		Text keyText=new Text();
		Text valueText=new Text();
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			//A	 I,K,C,B,G,F,H,O,D,
			String[] split = value.toString().split("\t");
			String friend=split[0];
			valueText.set(friend);
			String masters=split[1];
			String[] split2 = masters.split(",");
			Arrays.sort(split2);
			for (int i = 0; i < split2.length; i++) {
				String string = split2[i];
				for (int j = i+1; j < split2.length; j++) {
					String string2 = split2[j];
					keyText.set(string+"-"+string2);
					context.write(keyText, valueText);
				}
			}
		}
	}
	
	static class MutualFriendReduce extends Reducer<Text, Text, Text, Text>{
		Text valueText=new Text();
		@Override
		protected void reduce(Text key, Iterable<Text> iterable,Context context)
				throws IOException, InterruptedException {
			StringBuffer buffer=new StringBuffer();
			for (Text text : iterable) {
				buffer.append(text.toString()).append(",");
			}
			valueText.set(buffer.toString());
			context.write(key, valueText);
		}
		
	}
	
	public static void main(String[] args) throws Exception {
		Configuration configuration = new Configuration();
		Job job = Job.getInstance(configuration);
		job.setJarByClass(MutualFriendStepTwo.class);

		job.setMapperClass(MutualFriendMap.class);
		job.setReducerClass(MutualFriendReduce.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);

		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		boolean waitForCompletion = job.waitForCompletion(true);
		System.exit(waitForCompletion ? 0 : 1);
	}

}
